/**
 * 
 */

package com.wutianyi.nio.interfaces.impl;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.FutureTask;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.log4j.Logger;

import com.wutianyi.nio.interfaces.BufferFactory;
import com.wutianyi.nio.interfaces.ChannelFacade;
import com.wutianyi.nio.interfaces.Dispatcher;
import com.wutianyi.nio.interfaces.InputHandler;

/**
 * @author wutianyi
 * 
 */
public class NioDispatcher implements Dispatcher, Runnable {

	private final Logger logger = Logger.getLogger(NioDispatcher.class);
	private final ReadWriteLock selectorGuard = new ReentrantReadWriteLock();
	private final Executor executor;
	private final Selector selector;
	private final BlockingQueue<HandlerAdapter> statusChangeQueue;
	private final BufferFactory bufferFactory;
	private volatile boolean dispatching = true;

	public NioDispatcher(Executor executor, BufferFactory bufferFactory)
			throws IOException {

		this.executor = executor;
		this.bufferFactory = bufferFactory;
		statusChangeQueue = new ArrayBlockingQueue<HandlerAdapter>(100);
		selector = Selector.open();
	}

	public Thread start() {

		Thread thread = new Thread(this);
		thread.start();
		return thread;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.wutianyi.nio.interfaces.Dispatcher#registerChannel(java.nio.channels
	 * .SelectableChannel, com.wutianyi.nio.interfaces.InputHandler)
	 */
	@Override
	public ChannelFacade registerChannel(SelectableChannel channel,
			InputHandler handler) throws IOException {

		channel.configureBlocking(false);
		HandlerAdapter adapter = new HandlerAdapter(handler, this,
				bufferFactory);
		adapter.registering();
		acquireSelectorGuard();

		try {
			SelectionKey key = channel.register(selector, SelectionKey.OP_READ,
					adapter);
			adapter.setKey(key);
			adapter.registered();
			return adapter;
		} finally {
			releaseSelectorGuard();
		}
	}

	/**
	 * 
	 */
	private void releaseSelectorGuard() {

		selectorGuard.readLock().unlock();
	}

	/**
	 * 
	 */
	private void acquireSelectorGuard() {

		selectorGuard.readLock().lock();
		selector.wakeup();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.nio.interfaces.Dispatcher#shutdown()
	 */
	@Override
	public void shutdown() {

		dispatching = false;
		selector.wakeup();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * 
	 * @seecom.wutianyi.nio.interfaces.Dispatcher#unregisterChannel(com.wutianyi.
	 * nio.interfaces. ChannelFacade)
	 */
	@Override
	public void unregisterChannel(ChannelFacade token) {

		if (!(token instanceof HandlerAdapter)) {
			throw new IllegalArgumentException("Not a valid registration token");
		}
		HandlerAdapter adapter = (HandlerAdapter) token;
		SelectionKey selectionKey = adapter.key();
		acquireSelectorGuard();

		try {
			adapter.unregistering();
			selectionKey.cancel();
		} finally {
			releaseSelectorGuard();
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Runnable#run()
	 */
	@Override
	public void run() {

		try {
			dispatch();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}

		Set<SelectionKey> keys = selector.selectedKeys();
		for (SelectionKey key : keys) {
			HandlerAdapter adapter = (HandlerAdapter) key.attachment();
			unregisterChannel(adapter);
		}
		try {
			selector.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * @throws IOException
	 * 
	 */
	public void dispatch() throws IOException {

		while (dispatching) {
			selectorGuardBarrier();
			selector.select();
			checkStatusChangeQueue();
			Set<SelectionKey> keys = selector.selectedKeys();

			for (SelectionKey key : keys) {
				HandlerAdapter adapter = (HandlerAdapter) key.attachment();
				invokeHandler(adapter);
			}
			keys.clear();
		}
	}

	/**
	 * @param adapter
	 */
	private void invokeHandler(HandlerAdapter adapter) {

		adapter.prepareToRun();
		adapter.key().interestOps(0);
		executor.execute(new HandlerFutureTask(adapter));
	}

	/**
	 * 
	 */
	private void checkStatusChangeQueue() {

		HandlerAdapter adapter;

		while ((adapter = statusChangeQueue.poll()) != null) {
			if (adapter.isDead()) {
				unregisterChannel(adapter);
			} else {
				resumeSelection(adapter);
			}
		}
	}

	/**
	 * @param adapter
	 */
	private void resumeSelection(HandlerAdapter adapter) {

		SelectionKey key = adapter.key();
		if (key.isValid())
			key.interestOps(adapter.getInterestOps());
	}

	/**
	 * 
	 */
	private void selectorGuardBarrier() {

		selectorGuard.writeLock().lock();
		selectorGuard.writeLock().unlock();
	}

	void enqueueStatusChange(HandlerAdapter adapter) {

		boolean interrupted = false;

		try {
			while (true) {
				try {
					statusChangeQueue.put(adapter);
					selector.wakeup();
					return;
				} catch (InterruptedException e) {
					interrupted = true;
				}

			}
		} finally {
			if (interrupted)
				Thread.currentThread().interrupt();
		}
	}

	private class HandlerFutureTask extends FutureTask<HandlerAdapter> {

		private final HandlerAdapter adapter;

		public HandlerFutureTask(HandlerAdapter adapter) {

			super(adapter);
			this.adapter = adapter;
		}

		protected void done() {

			enqueueStatusChange(adapter);
			try {
				get();
			} catch (InterruptedException e) {
				Thread.interrupted();
				e.printStackTrace();
			} catch (ExecutionException e) {
				adapter.die();
				e.printStackTrace();
			}
		}
	}
}
